package com.xian.java.stream;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

public class TestSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setParallelism(1);

        DataStreamSource<String> dataStreamSource = env.addSource(new SourceFunction<String>() {

            String line = "-娇娇女神-";
            Random random = new Random();

            private long count = 0L;
            private volatile boolean isRunning = true;

            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                ctx.collect("自己定义源！");
                while (isRunning) {
                    count++;
                    ctx.collect(count+line+random.nextInt());
                    if (count==10000){
                        cancel();
                    }
                }
            }

            @Override
            public void cancel() {
                isRunning = false;
            }
        });


        dataStreamSource.print();


        env.execute();
    }
}
